/**
 * Copyright (C) 2017-2018 pinganfang, Inc. All Rights Reserved.
 */
package com.pingan.haofang.searchcloud.register.route;

import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import com.pingan.haofang.searchcloud.api.matedata.MetaCacheSupport;
import com.pingan.haofang.searchcloud.api.route.IndexRoute;
import org.apache.solr.common.cloud.SolrZkClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.Assert;

import java.util.Collections;
import java.util.concurrent.TimeUnit;

/**
 * CollectionRouteDiscover ZK默认实现
 * 
 * @author huhailiang937
 * @date 2017年4月7日下午4:45:16
 * @since 1.0.0
 */
public class Processor4ZKIndexDiscover {

    protected static Logger LOG = LoggerFactory.getLogger(Processor4ZKIndexDiscover.class);

    /**
     * 总大小：1024<br/>
     * 失效时间：写入之后的 60分3钟失效<br/>
     */
    protected static Cache<String, IndexRoute> registerCache = CacheBuilder.newBuilder().maximumSize(1024)
            .expireAfterWrite(3 * 60, TimeUnit.MINUTES).build();

    protected String metaZkHost;

    RouteChangeZKListener listener;

    public Processor4ZKIndexDiscover(String metaZkHost) {
        this.metaZkHost = metaZkHost;
        listener = new RouteChangeZKListener(metaZkHost);
        listener.start(registerCache);
    }

    public IndexRoute discover(String collectionRouteKey) {
        Assert.notNull(collectionRouteKey, "collectionRouteKey can not null");
        IndexRoute indexRoute = registerCache.getIfPresent(collectionRouteKey);
        if (null != indexRoute) {
            return indexRoute;
        }

        String routePath = MetaCacheSupport.getZKRoutePath(collectionRouteKey);
        SolrZkClient zkClient = null;
        try {
            zkClient = new SolrZkClient(metaZkHost, 30000);
            byte[] datas = zkClient.getData(routePath, null, null, false);
            IndexRoute routeInZk = MetaCacheSupport.decode(datas, IndexRoute.class);
            if (null != routeInZk.getEndpoints()) {
                Collections.sort(routeInZk.getEndpoints());
            }
            registerCache.put(collectionRouteKey, routeInZk);
            return routeInZk;
        } catch (Exception e) {
            LOG.error(String.format("ZKCollectionRouteDiscover discover(%s) has error: ", collectionRouteKey), e);
        } finally {
            if (null != zkClient) {
                zkClient.close();
            }
        }
        return null;
    }

}
